package trident.aggregation;

import org.apache.storm.trident.operation.BaseAggregator;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;

public class SumAggregation extends BaseAggregator<SumAggregation.SumState> {
    @Override
    public SumState init(Object batchId, TridentCollector collector) {
        return new SumState();
    }

    @Override
    public void aggregate(SumState val, TridentTuple tuple, TridentCollector collector) {
        Integer hive = tuple.getIntegerByField("hive");
        Integer storm = tuple.getIntegerByField("storm");

        val.hiveSum += hive;
        val.stormSum += storm;
    }

    @Override
    public void complete(SumState val, TridentCollector collector) {
        collector.emit(new Values(val));
    }

    static class SumState {
        Integer hiveSum = 0;
        Integer stormSum = 0;

        @Override
        public String toString() {
            return "SumState{" +
                    "hiveSum=" + hiveSum +
                    ", stormSum=" + stormSum +
                    '}';
        }
    }
}
